import os
import stat
import sys
import warnings
import tempfile312 as tempfile
import posixpath
from pathlib import PurePath
from dataclasses import dataclass
from typing import Dict, Iterable, Optional, Type, Union

from ofrak.core import GenericBinary

try:
    import xattr
except ImportError:
    import ofrak.core.xattr_stub as xattr  # type: ignore[no-redef]

from ofrak.model.viewable_tag_model import AttributesType
from ofrak.resource import Resource

from ofrak.model.resource_model import index, ResourceAttributes
from ofrak.model.tag_model import ResourceTag
from ofrak.resource_view import ResourceView
from ofrak.service.resource_service_i import (
    ResourceFilter,
    ResourceAttributeValueFilter,
    ResourceFilterCondition,
)


def _warn_chmod_chown_windows():  # pragma: no cover
    # warnings.warn instead of logging.warning prevents duplicating this static warning message
    warnings.warn(
        "os.chown and os.chmod do not work on Windows platforms. Unix-like file ownership and "
        "permissions will not be properly handled while using OFRAK on this platform."
    )


@dataclass
class FilesystemEntry(ResourceView):
    """
    Handles generic management of any entry stored within a filesystem.
    """

    name: str
    stat: Optional[os.stat_result]
    xattrs: Optional[Dict[str, bytes]]

    @index
    def Name(self) -> str:
        name = self.name.rstrip("/")
        return name.split("/")[-1]

    @classmethod
    def caption(cls, all_attributes) -> str:
        try:
            filesystem_attributes = all_attributes[AttributesType[FilesystemEntry]]
        except KeyError:
            return super().caption(all_attributes)
        return f"{cls.__name__}: {filesystem_attributes.name}"

    def get_name(self) -> str:
        """
        Get the base name of a folder.

        :return: The file or folder's base name
        """
        return self.Name

    async def set_stat(self, stat_result: os.stat_result):
        """
        Set the stat for the `FilesystemEntry`. Useful for newly created files where we want to
        control the full 10-tuple.

        :param stat_result: `os.stat_result` object containing all necessary values
        """
        self.stat = stat_result
        if self.resource is None:
            return
        all_view_attrs: Dict[
            Type[ResourceAttributes], ResourceAttributes
        ] = self.get_attributes_instances()
        filesystem_attrs = all_view_attrs[AttributesType[FilesystemEntry]]
        self.resource.add_attributes(filesystem_attrs)
        await self.resource.save()

    async def modify_stat_attribute(self, st_stat: int, stat_value: int):
        """
        Modify a specific `os.stat` attribute on the filesystem entry.

        Example:

        ```python
        fs_entry.modify_stat_attribute(stat.ST_MODE, 0o100755)
        ```

        :param st_stat: The `st_stat` struct member to be modified
        :param stat_value: The new stat value
        """
        if self.stat is None:
            raise ValueError("Cannot modify a stat attribute when stat attributes are not set")
        stat_attributes = list(self.stat)  # type: ignore
        stat_attributes[st_stat] = stat_value
        filesystem_stat_attributes = os.stat_result(tuple(stat_attributes))
        await self.set_stat(filesystem_stat_attributes)

    async def modify_xattr_attribute(self, attribute: str, value: bytes):
        """
        Modify the extended file attributes ("xattrs") for the filesystem entry.

        :param attribute:
        :param value:
        """
        if self.xattrs is None:
            self.xattrs = dict()
        self.xattrs[attribute] = value
        if self.resource is None:
            return
        await self.resource.save()

    async def get_path(self) -> str:
        """
        Get a folder's path, with the `FilesystemRoot` as the path root.

        :return: The full path name, with the `FilesystemRoot` ancestor as the path root;
        always POSIX style paths with forward slashes
        """
        path = [self.get_name()]

        for a in await self.resource.get_ancestors(
            r_filter=ResourceFilter(
                tags=(FilesystemEntry, FilesystemRoot),
                tags_condition=ResourceFilterCondition.OR,
            )
        ):
            if (a is None) or (a.has_tag(FilesystemRoot)):
                break
            a_view = await a.view_as(FilesystemEntry)
            path.append(a_view.get_name())

        return posixpath.join(*reversed(path))

    def apply_stat_attrs(self, path: str):
        """
        Set file mode and access times of a path on disk to match the attributes stored on this
        resource.

        :param path: Path on disk to set attributes of.
        """
        if self.stat:
            if sys.platform == "win32":
                _warn_chmod_chown_windows()
            else:
                os.chown(path, self.stat.st_uid, self.stat.st_gid)
                os.chmod(path, self.stat.st_mode)
            os.utime(path, (self.stat.st_atime, self.stat.st_mtime))
        if self.xattrs:
            for attr, value in self.xattrs.items():
                xattr.setxattr(path, attr, value)

    def is_file(self) -> bool:
        return self.resource.has_tag(File)

    def is_folder(self) -> bool:
        return self.resource.has_tag(Folder)

    def is_link(self) -> bool:
        return self.resource.has_tag(SymbolicLink)

    def is_block_device(self) -> bool:
        return self.resource.has_tag(BlockDevice)

    def is_fifo_pipe(self) -> bool:
        return self.resource.has_tag(FIFOPipe)

    def is_character_device(self) -> bool:
        return self.resource.has_tag(CharacterDevice)

    def is_device(self) -> bool:
        return self.is_block_device() or self.is_character_device()

    async def flush_to_disk(self, root_path: str = ".", filename: Optional[str] = None):
        entry_path = await self.get_path()
        if filename is not None:
            entry_path = os.path.join(os.path.dirname(entry_path), filename)
        if self.is_link():
            link_name = os.path.join(root_path, entry_path)
            if not os.path.exists(link_name):
                link_view = await self.resource.view_as(SymbolicLink)
                os.symlink(link_view.source_path, link_name)
            assert len(list(await self.resource.get_children())) == 0
            if self.stat:
                if sys.platform == "win32":
                    _warn_chmod_chown_windows()
                else:
                    # https://docs.python.org/3/library/os.html#os.supports_follow_symlinks
                    if os.chown in os.supports_follow_symlinks:
                        os.chown(
                            link_name, self.stat.st_uid, self.stat.st_gid, follow_symlinks=False
                        )
                    if os.chmod in os.supports_follow_symlinks:
                        os.chmod(link_name, self.stat.st_mode, follow_symlinks=False)
                if os.utime in os.supports_follow_symlinks:
                    os.utime(
                        link_name,
                        (self.stat.st_atime, self.stat.st_mtime),
                        follow_symlinks=False,
                    )
            if self.xattrs:
                for attr, value in self.xattrs.items():
                    xattr.setxattr(link_name, attr, value, symlink=True)  # Don't follow links
        elif self.is_folder():
            folder_name = os.path.join(root_path, entry_path)
            if not os.path.exists(folder_name):
                os.makedirs(folder_name)
        elif self.is_file():
            file_name = os.path.join(root_path, entry_path)
            with open(file_name, "wb") as f:
                f.write(await self.resource.get_data())
            self.apply_stat_attrs(file_name)
        elif self.is_device():
            device_name = os.path.join(root_path, entry_path)
            if sys.platform == "win32" or not hasattr(os, "mknod"):
                warnings.warn(
                    f"Cannot create a device {entry_path} for a "
                    f"BlockDevice or CharacterDevice resource on platform {sys.platform}! "
                    f"Creating an empty regular file instead."
                )
                with open(device_name, "w") as f:
                    pass
            else:
                if self.stat is None:
                    raise ValueError(
                        f"Cannot create a device {entry_path} for a "
                        f"BlockDevice or CharacterDevice resource with no stat!"
                    )
                os.mknod(device_name, self.stat.st_mode, self.stat.st_rdev)

            self.apply_stat_attrs(device_name)
        elif self.is_fifo_pipe():
            fifo_name = os.path.join(root_path, entry_path)
            if sys.platform == "win32" or not hasattr(os, "mkfifo"):
                warnings.warn(
                    f"Cannot create a fifo {entry_path} for a "
                    f"FIFOPipe resource on platform {sys.platform}! "
                    f"Creating an empty regular file instead."
                )
                with open(fifo_name, "w") as f:
                    pass
            else:
                if self.stat is None:
                    raise ValueError(
                        f"Cannot create a fifo {entry_path} for a FIFOPipe resource "
                        "with no stat!"
                    )
                os.mkfifo(fifo_name, self.stat.st_mode)

            self.apply_stat_attrs(fifo_name)
        else:
            entry_info = f"Stat: {stat.S_IFMT(self.stat.st_mode):o}" if self.stat else ""
            raise NotImplementedError(
                f"FilesystemEntry {entry_path} has an unknown or "
                f"unsupported filesystem type! Unable to create it "
                f"on-disk. {entry_info}"
            )


class File(FilesystemEntry, GenericBinary):
    """
    Stores the data and location of a file within a filesystem or folder's descendant file tree.
    """


class Folder(FilesystemEntry):
    """
    Describes a folder that is stored in a filesystem as a file tree.

    All descendant resources are stored in a tree structure that reflects a folder/directory's file
    tree.
    """

    async def get_entry(self, path: str) -> Optional[FilesystemEntry]:
        """
        Search a folder for an entry with the given path.

        :param path: The filesystem path to search for, relative to this folder

        :return: The child `FilesystemEntry` resource that was found. If nothing was found, `None`
        is returned
        """
        basename = posixpath.basename(path)
        # only searching paths with the same base name should reduce the search space by quite a lot
        descendants = await self.resource.get_descendants_as_view(
            FilesystemEntry,
            r_filter=ResourceFilter(
                attribute_filters=(ResourceAttributeValueFilter(FilesystemEntry.Name, basename),)
            ),
        )

        for d in descendants:
            descendant_path = await d.get_path()
            if descendant_path.split(f"{self.name}/")[-1] == path:
                return d
        return None

    async def list_dir(self) -> Dict[str, FilesystemEntry]:
        """
        Enumerate a folder's children, much like `os.listdir`.

        :return: A dictionary of child entries, with the child's name as the key
        """
        entries = dict()
        for c in await self.resource.get_children_as_view(FilesystemEntry):
            entries[c.get_name()] = c
        return entries


@dataclass
class SpecialFileType(FilesystemEntry):
    """
    A filesystem entry that is not a simple type, like a file or folder. For example, symbolic
    links and block devices fall under this category.
    """


@dataclass
class SymbolicLink(SpecialFileType):
    """
    Symbolic link pointing to a (possibly invalid) path in the filesystem. The path pointed to is
    invalid if accessing it would raise a `FileNotFoundError`.

    :ivar source_path: File pointed to by the symbolic link
    """

    source_path: str


@dataclass
class BlockDevice(SpecialFileType):
    """
    Block special device file.
    """


@dataclass
class FIFOPipe(SpecialFileType):
    """
    Named pipe.
    """


@dataclass
class CharacterDevice(SpecialFileType):
    """
    Character special device file.
    """


def _path_to_posixpath(path: str) -> str:
    """
    Converts an OS-specific path to a POSIX style path
    """
    return str(PurePath(path).as_posix())


@dataclass
class FilesystemRoot(ResourceView):
    """
    A resource that contains a filesystem's file tree. All descendant resources are stored in a tree
    structure that reflects a filesystem's file tree. The methods within this class are intended to
    be used as utilities when unpacking any filesystem-like resource.

    Any resource that contains a file tree should inherit the `FilesystemRoot` class.
    """

    async def initialize_from_disk(
        self,
        path: str,
    ):
        root_path = os.path.normpath(path)

        if not os.path.exists(root_path):
            raise FileNotFoundError(
                f"Could not initialize from disk. Root path does not exist: {root_path}"
            )
        if not os.path.isdir(root_path):
            raise FileNotFoundError(
                f"Could not initialize from disk. Found a file instead of a directory: {root_path}"
            )

        for root, dirs, files in os.walk(root_path):
            for d in sorted(dirs):
                absolute_path = os.path.join(root, d)
                relative_path_posix = posixpath.join(
                    _path_to_posixpath(os.path.relpath(root, root_path)), d
                )

                folder_attributes_stat = os.lstat(absolute_path)

                mode = folder_attributes_stat.st_mode
                mode_tests = [
                    stat.S_ISCHR,
                    stat.S_ISBLK,
                    stat.S_ISFIFO,
                    stat.S_ISSOCK,
                    stat.S_ISDOOR,
                    stat.S_ISPORT,
                    stat.S_ISWHT,
                    stat.S_ISREG,
                ]
                for mode_test in mode_tests:
                    if mode_test(mode) != 0:
                        raise NotImplementedError(
                            f"Directory {absolute_path} has an unsupported special file type: "
                            f"{stat.S_IFMT(mode):o}. {mode_test.__name__} should be false."
                        )

                folder_attributes_xattr = self._get_xattr_map(absolute_path)
                if os.path.islink(absolute_path):
                    await self.add_special_file_entry(
                        relative_path_posix,
                        SymbolicLink(
                            relative_path_posix,
                            folder_attributes_stat,
                            folder_attributes_xattr,
                            os.readlink(absolute_path),
                        ),
                    )
                else:
                    await self.add_folder(
                        relative_path_posix,
                        folder_attributes_stat,
                        folder_attributes_xattr,
                    )
            for f in sorted(files):
                absolute_path = os.path.join(root, f)
                relative_path_posix = _path_to_posixpath(
                    os.path.normpath(os.path.join(os.path.relpath(root, root_path), f))
                )
                file_attributes_stat = os.lstat(absolute_path)

                mode = file_attributes_stat.st_mode
                mode_tests = [
                    stat.S_ISSOCK,
                    stat.S_ISDOOR,
                    stat.S_ISPORT,
                    stat.S_ISWHT,
                    stat.S_ISDIR,
                ]
                for mode_test in mode_tests:
                    if mode_test(mode) != 0:
                        raise NotImplementedError(
                            f"Directory {absolute_path} has an unsupported special file type: "
                            f"{stat.S_IFMT(mode):o}. {mode_test.__name__} should be false."
                        )

                file_attributes_xattr = self._get_xattr_map(absolute_path)
                if os.path.islink(absolute_path):
                    await self.add_special_file_entry(
                        relative_path_posix,
                        SymbolicLink(
                            relative_path_posix,
                            file_attributes_stat,
                            file_attributes_xattr,
                            os.readlink(absolute_path),
                        ),
                    )
                elif os.path.isfile(absolute_path):
                    with open(absolute_path, "rb") as fh:
                        await self.add_file(
                            relative_path_posix,
                            fh.read(),
                            file_attributes_stat,
                            file_attributes_xattr,
                        )
                elif stat.S_ISFIFO(mode):
                    await self.add_special_file_entry(
                        relative_path_posix,
                        FIFOPipe(relative_path_posix, file_attributes_stat, file_attributes_xattr),
                    )
                elif stat.S_ISBLK(mode):
                    await self.add_special_file_entry(
                        relative_path_posix,
                        BlockDevice(
                            relative_path_posix, file_attributes_stat, file_attributes_xattr
                        ),
                    )
                elif stat.S_ISCHR(mode):
                    await self.add_special_file_entry(
                        relative_path_posix,
                        CharacterDevice(
                            relative_path_posix, file_attributes_stat, file_attributes_xattr
                        ),
                    )
                else:
                    raise NotImplementedError(
                        f"File {absolute_path} appeared to be a supported "
                        f"type but did not match any of the known cases to "
                        f"create a resource. Stat: {stat.S_IFMT(mode):o}"
                    )

    async def flush_to_disk(
        self,
        path: Optional[str] = None,
    ):
        """
        Writes this `FilesystemRoot`'s `FilesystemEntry` descendants to directory. If a target path
        is not provided, the output is written to a temporary directory.

        :return: the root directory containing the flushed filesystem
        """
        if path is None:
            root_path = tempfile.mkdtemp()
        else:
            root_path = path

        entries = [
            f
            for f in await self.resource.get_children_as_view(
                FilesystemEntry, r_filter=ResourceFilter(tags=(FilesystemEntry,))
            )
        ]
        while len(entries) > 0:
            entry = entries.pop(0)
            if entry.is_folder():
                for child in await entry.resource.get_children_as_view(
                    FilesystemEntry, r_filter=ResourceFilter(tags=(FilesystemEntry,))
                ):
                    entries.append(child)
            await entry.flush_to_disk(root_path=root_path)

        return root_path

    async def get_entry(self, path: str):
        """
        Searches this `FilesystemRoot`'s descendants for a filesystem entry with a given path,
            and returns that entry if found.

        :param path: the path of the `FilesystemEntry` to search for, with this `FilesystemRoot` as
        the path root

        :return: the descendant `FilesystemEntry`, if found; otherwise, returns `None`
        """
        basename = posixpath.basename(path)
        # only searching paths with the same base name should reduce the search space by quite a lot
        descendants = await self.resource.get_descendants_as_view(
            FilesystemEntry,
            r_filter=ResourceFilter(
                attribute_filters=(ResourceAttributeValueFilter(FilesystemEntry.Name, basename),)
            ),
        )

        for d in descendants:
            if await d.get_path() == posixpath.normpath(path):
                return d
        return None

    async def list_dir(self) -> dict:
        """
        Enumerates a `FilesystemRoot`'s children, much like `os.listdir`.

        :return: a dictionary of child entries, with the child's name as the key
        """
        entries = dict()
        for c in await self.resource.get_children_as_view(FilesystemEntry):
            entries[c.get_name()] = c
        return entries

    async def add_folder(
        self,
        path: str,
        folder_stat_result: Optional[os.stat_result] = None,
        folder_xattrs: Optional[Dict[str, bytes]] = None,
        tags: Iterable[ResourceTag] = (),
        attributes: Iterable[ResourceAttributes] = (),
    ) -> Folder:
        """
        Adds a [Folder][ofrak.core.filesystem.Folder] resource to a `FilesystemRoot`, creating all
        parent folders as needed.

        :param path: the path that will contain the folder to be added
        :param folder_stat_result: the filesystem attributes associated with the folder
        :param folder_xattrs: xattrs for the folder
        :param tags: the list of tags to be added to the new resource. The `Folder` tag is added by
        default
        :param attributes: the list of additional attributes to be added to the new folder, the
        folder's name attribute is added automatically

        :raises ValueError: if the path is too short and doesn't actually include any directories

        :return: the `Folder` resource that was added to the `FilesystemRoot`
        """
        # Normalizes and cleans up paths beginning with "./" and containing "./../" as well as
        # other extraneous separators
        split_dir = posixpath.normpath(path).strip("/").split("/")

        parent: Union[FilesystemRoot, Folder] = self
        for directory in split_dir:
            folder_entries = await parent.list_dir()

            if directory not in folder_entries.keys():
                new_missing_folder = await parent.resource.create_child_from_view(
                    Folder(directory, folder_stat_result, folder_xattrs),
                    data=b"",
                    additional_tags=tags,
                    additional_attributes=attributes,
                )
                parent = await new_missing_folder.view_as(Folder)
            else:
                parent = await folder_entries[directory].resource.view_as(Folder)

        if type(parent) is FilesystemRoot:
            assert len(split_dir) == 0  # Only case this should happen
            raise ValueError(f"The path {path} is too short (no directories)")

        if not isinstance(parent, Folder):
            raise ValueError(
                f"Parent folder {parent} has an unexpected type {type(parent)}. It "
                f"should be a Folder instead."
            )
        return parent

    async def add_file(
        self,
        path: str,
        data: bytes,
        file_stat_result: Optional[os.stat_result] = None,
        file_xattrs: Optional[Dict[str, bytes]] = None,
        tags: Iterable[ResourceTag] = (),
        attributes: Iterable[ResourceAttributes] = (),
    ) -> Resource:
        """
        Adds a [File][ofrak.core.filesystem.File] resource to a `FilesystemRoot`, creating all
        parent [Folders][ofrak.core.filesystem.Folder] as needed.

        :param path: the path that will contain the `File` to be added
        :param data: contents of the file being added
        :param file_stat_result: the filesystem attributes associated with the file
        :param file_xattrs: xattrs for the file
        :param tags: the list of tags to be added to the new resource, the File tag is added by
        default
        :param attributes: the list of additional attributes to be added to the new Folder,
            the file's name attribute is added automatically

        :return: the `File` resource that was added to the `FilesystemRoot`
        """
        dirname = posixpath.dirname(path)
        filename = posixpath.basename(path)

        if dirname == "":
            parent_folder = self
        else:
            parent_folder = await self.get_entry(dirname)
            if parent_folder is None:
                parent_folder = await self.add_folder(dirname)

        new_file = await parent_folder.resource.create_child_from_view(
            File(filename, file_stat_result, file_xattrs),
            data=data,
            additional_tags=tags,
            additional_attributes=attributes,
        )
        return new_file

    async def remove_file(self, path: str) -> None:
        """
        Removes a [File][ofrak.core.filesystem.File] resource from a `FilesystemRoot`.
        :param path: the path of the file to be removed
        :return: None
        """
        file_to_remove = await self.get_entry(path)
        await file_to_remove.resource.delete()
        await file_to_remove.resource.save()

    async def add_special_file_entry(
        self,
        path: str,
        special_file_view: SpecialFileType,
        tags: Iterable[ResourceTag] = (),
        attributes: Iterable[ResourceAttributes] = (),
    ) -> Resource:
        """
        Adds a resource representing a [SpecialFileType][ofrak.core.filesystem.SpecialFileType]
        to a `FilesystemRoot`, creating all parent [Folders][ofrak.core.filesystem.Folder] as
        needed.

        Some examples of these "special" types are
        [SymbolicLink][ofrak.core.filesystem.SymbolicLink] and
        [BlockDevice][ofrak.core.filesystem.BlockDevice].

        :param path: The path of the `FilesystemEntry` to be added
        :param special_file_view: A ResourceView, whose type should be a subclass of
        `FilesystemEntry`
        :param tags: the list of tags to be added to the new resource, the File tag is added by
        default
        :param attributes: the list of additional attributes to be added to the new Folder,
        the file's name attribute is added automatically

        :return: The special `FilesystemEntry` resource that was added to the `FilesystemRoot`
        """
        dirname = posixpath.dirname(path)

        if dirname == "":
            parent_folder = self
        else:
            parent_folder = await self.get_entry(dirname)
            if parent_folder is None:
                parent_folder = await self.add_folder(dirname)

        new_entry = await parent_folder.resource.create_child_from_view(
            special_file_view,
            data=b"",  # Use empty, non-None data
            additional_tags=tags,
            additional_attributes=attributes,
        )
        return new_entry

    @classmethod
    def _get_xattr_map(cls, path):
        xattr_dict = {}
        for attr in xattr.listxattr(path, symlink=True):  # Don't follow links
            xattr_dict[attr] = xattr.getxattr(path, attr)
        return xattr_dict
